package com.cyx.netty.gateway.client.impl;
/*
 * Copyright 2013-2017 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */
import com.alibaba.fastjson.JSON;
import com.cyx.netty.gateway.client.NettyRouterTransportService;
import com.cyx.netty.gateway.GatewayNettyApplication;
import com.cyx.netty.gateway.client.channel.ProxyClientChannelInitializer;
import com.cyx.netty.gateway.config.NettyEventThreadConfig;
import com.cyx.netty.gateway.route.Route;
import com.cyx.netty.gateway.route.RouteLocator;
import com.cyx.netty.gateway.support.ServerWebExchange;
import com.cyx.netty.gateway.util.UriComponentsBuilder;
import com.cyx.netty.gateway.util.cache.ChannelCacheManager;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;

import java.net.URI;
import java.util.List;
import java.util.Random;

import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
import static io.netty.handler.codec.http.HttpHeaderValues.KEEP_ALIVE;
import static io.netty.handler.codec.http.HttpHeaders.Names.*;
import static io.netty.handler.codec.http.HttpHeaders.Names.CONNECTION;
import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;

@Service
public class NettyRouterTransportImpl implements NettyRouterTransportService {

  private final Logger logger = LoggerFactory.getLogger(NettyRouterTransportImpl.class);

  private NettyEventThreadConfig nettyEventThreadConfig;
  private GatewayNettyApplication springNettyApplication;
  //  final WeightRoundRobin r = new WeightRoundRobin();
  private static   Random rand = new Random();

  public NettyRouterTransportImpl(NettyEventThreadConfig nettyEventThreadConfig,
      GatewayNettyApplication springNettyApplication) {
    this.nettyEventThreadConfig = nettyEventThreadConfig;
    this.springNettyApplication=springNettyApplication;
  }


  @Override
  public void transport(ServerWebExchange exchange) {

    //重连至目标服务器
//    Channel channel = ChannelCacheManager.get(exchange.getChannelHandlerContext().channel().id().asShortText());
//    if (channel!=null && channel.isActive()){
//
//        logger.info("replay channel >>");
//        channel.writeAndFlush(exchange.getRequest());
//        return;
//    }

    RouteLocator routeLocator = springNettyApplication.customRouteLocator();
    Mono<List<Route>> routeMono = routeLocator.getRoutes().filter(r -> r.getPredicate().test(exchange)).collectList();
    List<Route> block = routeMono.block();
    if (block==null || block.isEmpty()){
      logger.info("not predicate one");

      ByteBuf netty_test = Unpooled.copiedBuffer("{\"success\":\"fail\",\"msg\":\"谓词不匹配,无法找到相关服务\"}", CharsetUtil.UTF_8);


      Channel channel = exchange.getChannelHandlerContext().channel();

      FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1,HttpResponseStatus.OK,netty_test);
      response.headers().set(ACCESS_CONTROL_ALLOW_ORIGIN, "*");
      response.headers().set(ACCESS_CONTROL_ALLOW_HEADERS,"*");
      response.headers().set(ACCESS_CONTROL_ALLOW_CREDENTIALS, "true");
      response.headers().setInt(CONTENT_LENGTH, netty_test.readableBytes());

      boolean keepAlive = HttpUtil.isKeepAlive(response);
      if (!keepAlive) {
        channel.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
        return;
      }
      response.headers().set(CONNECTION, KEEP_ALIVE);
      channel.writeAndFlush(response);

      return;
    }

    Bootstrap  bootstrap = getBootstrap();
    bootstrap.handler(new ProxyClientChannelInitializer(exchange));

    Route route = block.get(rand.nextInt(block.size()));
    URI toUri = UriComponentsBuilder.fromUriString(exchange.getRequest().uri())
        .uri(route.getUri()).build(true).toUri();

    connect(exchange, toUri, bootstrap);
  }

  private  Bootstrap getBootstrap() {
    return new Bootstrap()
            .group(nettyEventThreadConfig.workerGroup())
            .channel(NioSocketChannel.class)
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);
  }

  private void connect(ServerWebExchange exchange, URI uri,
      Bootstrap bootstrap) {

    ChannelFuture cf = bootstrap.connect(uri.getHost(), uri.getPort());
    cf.addListener((ChannelFutureListener) future -> {
      if (future.isSuccess()) {
        future.channel().writeAndFlush(exchange.getRequest());
      } else {

        logger.info("connect fail \n");
        exchange.getChannelHandlerContext().channel().writeAndFlush(
            new DefaultFullHttpResponse(HTTP_1_1, NOT_FOUND, Unpooled.wrappedBuffer( JSON.toJSONBytes("connect fail"))));
        exchange.getChannelHandlerContext().channel().close();
      }
    });
  }

}

//    logger.info("打印请求地址："+build.toString());
//    bootstrap.channel(NioSocketChannel.class);
//     bootstrap.channel(ctx.channel().getClass()); // 使用NioSocketChannel来作为连接用的channel类
//    Server round = r.round();
//    logger.info("打印请求地址："+round.getIp()+":"+round.getPort());

//    Map<String, String> addressMap = new HashMap<>();
//    addressMap.put("host", s[0]);
//    addressMap.put("port", s[1]);

//        stringRedisTemplate.boundHashOps(key).putAll(addressMap);
//        stringRedisTemplate.boundHashOps(key).expire(30L, TimeUnit.SECONDS);

//        String key = msg.headers().get("key");
//        Map<Object, Object> address =  stringRedisTemplate.boundHashOps(key).entries();
//        String host = (String) address.get("host");
//        int port = Integer.parseInt((String)address.get("port"));

//    String host = "127.0.0.1";
//    int port = 8977;
//        if (map.get(host+"/"+port)==null){
